package api.flinkapi.apidemos;

import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.DoubleCounter;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.Serializable;
import java.util.Locale;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/7/26
 **/
public class GuageDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        DataStreamSource<String> ds = env.socketTextStream("localhost", 5656);
        KeyedStream<String, String> keyd = ds.keyBy(s -> s.split(",")[0]);

        SingleOutputStreamOperator<String> map = keyd.map(new RichMapFunction<String, String>() {
            Accumulator<String, Integer> acc;
            IntCounter intCnt;
            DoubleCounter dbCnt;
            LongCounter lngCnt;
            Counter g1_c1;
            Counter g1_c2;
            Counter g2_c1;
            int gauge;
            org.apache.flink.api.common.accumulators.IntCounter accIntCnt= new org.apache.flink.api.common.accumulators.IntCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                //accIntCnt = new org.apache.flink.api.common.accumulators.IntCounter();
                getRuntimeContext().addAccumulator("acc",this.accIntCnt);

                intCnt = getRuntimeContext().getIntCounter("intCnt");
                dbCnt = getRuntimeContext().getDoubleCounter("dbCnt");
                lngCnt = getRuntimeContext().getLongCounter("lngCnt");
                MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
                MetricGroup g1 = metricGroup.addGroup("g1");
                g1_c1 = g1.counter("g1_c1");
                g1_c2 = g1.counter("g1_c2");
                MetricGroup g2 = metricGroup.addGroup("g2");
                g2_c1 = g2.counter("g2_c1");
                g2.gauge("g2_gauge1", new Gauge<Integer>() {
                    @Override
                    public Integer getValue() {
                        return gauge;
                    }
                });
            }

            @Override
            public String map(String value) throws Exception {
                this.accIntCnt.add(1);
                intCnt.add(1);
                dbCnt.add(2.0);
                lngCnt.add(3);
                g1_c1.inc(4);
                g1_c2.inc(5);
                g2_c1.inc(6);
                gauge += 7;

                return value.toUpperCase();
            }
        });

        map.print();
        env.execute();

    }


}
